package com.mai.realtime.app.dwd.db;

import com.mai.realtime.app.BaseSqlApp;
import com.mai.realtime.common.Constant;
import com.mai.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * com.mai.realtime.app.dwd.db.Dwd_CourseTestExam
 * 考试情况事实表
 */
public class Dwd_CourseTestExam extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_CourseTestExam().init(
                3005,
                3,
                "Dwd_CourseTestExam"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        // 关联状态保存时间TTL
        tEnv.getConfig().setIdleStateRetention(Duration.ofMinutes(10));

        //读取ods_db
        readOdsDb(tEnv,"Dwd_CourseTestExam");

        // 从动态表读取测验表
        Table testExam = tEnv.sqlQuery("select " +
                " data['id'] id , " +//考试id
                " data['paper_id'] paper_id ," +//纸张编号
                " data['user_id'] user_id , " +//考生id
                " data['score'] score , " +//分数
                " data['duration_sec'] duration_sec ," +//用时
                " data['submit_time'] submit_time ," +//交卷时间
                " data['create_time'] create_time ," +//开始时间
                " data['update_time'] update_time ," +//更新时间
                " ts" +
                " from ods_db" +
                " where `database`='gmall' " +
                " and `table`='test_exam' " +
                " and `type`='insert'");

        tEnv.createTemporaryView("test_exam",testExam);

        // 从动态表读取测验问题表
        Table testPaper = tEnv.sqlQuery("select " +
                " data['id'] id , " +
                " data['paper_title'] paper_title ," +
                " data['course_id'] course_id , " +
                " data['create_time'] create_time , " +
                " data['update_time'] update_time ," +
                " data['publisher_id'] publisher_id ," +
                " data['deleted'] deleted ," +
                " from ods_db" +
                " where `database`='gmall' " +
                " and `table`='test_paper' " +
                " and `type`='insert'");

        tEnv.createTemporaryView("testPaper",testPaper);

        //两表join
        Table result = tEnv.sqlQuery("select " +
                " t1.id," +
                " t1.paper_id," +
                " t1.user_id," +
                " t1.score," +
                " t1.duration_sec," +
                " t1.create_time," +
                " t1.update_time," +
                " t1.submit_time," +
                " t2.course_id," +
                " t1.ts " +
                " from test_exam t1" +
                " join testPaper t2" +
                " on  t1.`paper_id`=t2.`id`" +
                "");

        // 定义动态表输出topic关联
        tEnv.executeSql("create table dwd_course_test_exam(" +
                " id string," +
                " paper_id string," +
                " user_id string," +
                " score string," +
                " duration_sec string," +
                " create_time string," +
                " update_time string," +
                " submit_time string," +
                " course_id string," +
                " ts BIGINT" +
                ")"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_COURSE_TEST_EXAM)
        );

        // 将结果输出DWD
        result.executeInsert("dwd_course_test_exam");
        //result.execute().print();



    }
}
